package net.bw.realtime.tmall.ods.job;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.github.javafaker.Faker;
import net.bw.realtime.tmall.common.utils.KafkaUtil;
import net.bw.realtime.tmall.ods.bean.TmallLogBean;
import net.bw.realtime.tmall.ods.utils.RandomTimestampToday;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.util.*;
import java.util.concurrent.TimeUnit;

import static net.bw.realtime.tmall.ods.utils.ChineseLocationUtils.getRandomCity;
import static net.bw.realtime.tmall.ods.utils.ChineseLocationUtils.getRandomProvince;

public class AbnormalTmallLogData {

	public static void main(String[] args) throws Exception {
		// 创建执行环境
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1);

		DataStreamSource<TmallLogBean> tmallLogBeanDataStreamSource = env.addSource(
				new RichParallelSourceFunction<TmallLogBean>() {

					private volatile boolean isRunning = true;
					private final Random random = new Random();
					private final String[] actionTypes = {"view_product", "add_to_cart", "place_order", "pay_success"};
					private final String[] categories = {"手机数码", "家用电器", "服饰箱包", "美妆护肤", "食品饮料"};
					private final String[] deviceTypes = {"PC", "无线"};

					// 6个特定用户ID
					private final String[] specialUsers = {
							"user_888001", "user_888002", "user_888003",
							"user_888004", "user_888005", "user_888006"
					};

					// 记录每个用户的大额订单次数
					private final Map<String, Integer> bigOrderCount = new HashMap<>();

					// 记录是否已经生成过大额订单
					private boolean bigOrdersGenerated = false;

					@Override
					public void run(SourceContext<TmallLogBean> ctx) throws Exception {
						int count = 0;
						long startTime = System.currentTimeMillis();

						while (isRunning && count < 10000) {
							TmallLogBean tmallLogBean = new TmallLogBean();

							// 设置时间戳 - 确保大额订单在一分钟内生成
							long timestamp;
							if (!bigOrdersGenerated && count < 18) { // 6用户×3订单=18
								// 在一分钟内生成大额订单
								timestamp = startTime + random.nextInt(60000);
							} else {
								timestamp = new RandomTimestampToday().getRandomTimestampToday();
							}
							tmallLogBean.setTs(timestamp);

							// 用户ID - 前18个订单使用6个特定用户
							String currentUserId;
							if (count < 18) {
								currentUserId = specialUsers[count % 6];
							} else {
								currentUserId = "user_" + (100000 + random.nextInt(900000));
							}
							tmallLogBean.setUserId(currentUserId);

							// 订单ID
							tmallLogBean.setOrderId("order_"+ random.nextInt(1000000));
							tmallLogBean.setUserName(Faker.instance().name().fullName());

							// 行为类型 - 前18个订单为支付成功
							String actionType;
							if (count < 18) {
								actionType = "pay_success";
								bigOrderCount.put(currentUserId, bigOrderCount.getOrDefault(currentUserId, 0) + 1);
							} else {
								actionType = actionTypes[random.nextInt(actionTypes.length)];
							}
							tmallLogBean.setActionType(actionType);

							// 商品详情
							Map<String, Object> actionDetail = new HashMap<>();
							actionDetail.put("shop_id", "shop_" + (10000 + random.nextInt(90000)));
							actionDetail.put("shop_name", Faker.instance().company().name());
							actionDetail.put("category", categories[random.nextInt(categories.length)]);
							actionDetail.put("brand", Faker.instance().company().name());

							// 价格 - 前18个订单设置为10000元以上
							double price;
							if (count < 18) {
								price = 10000 + random.nextDouble() * 50000; // 10000-60000元
							} else {
								price = 10 + random.nextDouble() * 10000; // 10-10010元
							}
							actionDetail.put("price", price);

							actionDetail.put("skuId", "sku_" + (10000 + random.nextInt(90000)));
							actionDetail.put("sku_name", Faker.instance().commerce().productName());
							actionDetail.put("quantity", 1 + random.nextInt(10)); // 1-10件
							tmallLogBean.setActionDetail(actionDetail);

							// 设备信息
							Map<String, String> deviceInfo = new HashMap<>();
							deviceInfo.put("deviceId", "device_" + random.nextInt(100000));
							deviceInfo.put("deviceType", deviceTypes[random.nextInt(deviceTypes.length)]);
							deviceInfo.put("os", "Android 12");
							deviceInfo.put("osVersion", "1.0.0");
							deviceInfo.put("appVersion", "Tmall_10.15.0");
							tmallLogBean.setDeviceInfo(deviceInfo);

							// 位置信息
							Faker fakerZh = Faker.instance(new Locale("zh-CN"));
							Map<String, String> location = new HashMap<>();
							location.put("ip", fakerZh.internet().ipV4Address());
							location.put("country", "中国");
							String province = getRandomProvince();
							String city = getRandomCity(province);
							location.put("province", province);
							location.put("city", city);
							tmallLogBean.setLocation(location);

							// 会话信息
							Map<String, String> session = new HashMap<>();
							session.put("sessionId", "session_" + System.currentTimeMillis());
							session.put("referer", "https://s.click.taobao.com/xxxxxx");
							session.put("pageUrl", "https://detail.tmall.com/item.htm?id=" + (10000 + random.nextInt(90000)));
							tmallLogBean.setSession(session);

							ctx.collect(tmallLogBean);
							count++;

							if (count >= 18 && !bigOrdersGenerated) {
								bigOrdersGenerated = true;
								System.out.println("已生成6个用户的大额订单数据，每个用户3笔订单");
								printBigOrderStats();
							}

							Thread.sleep(10); // 控制数据生成速度
						}
					}

					// 打印大额订单统计信息
					private void printBigOrderStats() {
						System.out.println("========== 大额订单统计 ==========");
						for (String userId : specialUsers) {
							System.out.printf("用户 %s: %d 笔大额订单%n", userId, bigOrderCount.getOrDefault(userId, 0));
						}
						System.out.println("===============================");
					}

					@Override
					public void cancel() {
						isRunning = false;
					}
				}
		);

		ObjectMapper mapper = new ObjectMapper();

		KafkaUtil.producerKafka(tmallLogBeanDataStreamSource.map(mapper::writeValueAsString), "tmall_log_test_abnormal");

		env.execute("TmallLogMockJob");
	}
}